package com.zeny.express.framework.mq.activemq;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQBytesMessage;

public class ActiveMqQueueHandler {
	
	private String userName;
	private String password;
	private String brokerURL;
	private String queueName;
	
	private Connection connection;
	private Session session;
	private Destination destination;
	private MessageProducer producer;
	private MessageConsumer consumer;
	
	public ActiveMqQueueHandler(String userName, String password, String brokerURL,
            String queueName) {
        super();

        this.userName = userName;
        this.password = password;
        this.brokerURL = brokerURL;
        this.queueName = queueName;
    }
	
	public void make() throws Exception {
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                userName, password, brokerURL);
    	connection = connectionFactory.createConnection();
    	connection.start();
		session = connection.createSession(Boolean.TRUE,
                Session.AUTO_ACKNOWLEDGE);
        destination = session.createQueue(queueName);
        //TODO:设置为可配选项
        
        producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    }
	
	public void close() throws Exception {
		session.close();
		connection.close();
	}
	
	public void commit() throws JMSException
	{
		session.commit();
	}
	
	public void rollback() throws JMSException
	{
		session.rollback();
	}
	
	public void send(byte[] bytes) throws JMSException
	{
		BytesMessage message = session.createBytesMessage();
		message.writeBytes(bytes);
		producer.send(message);
	}
	
	public byte[] receive() throws JMSException
	{
        consumer = session.createConsumer(destination);
		ActiveMQBytesMessage message = (ActiveMQBytesMessage) consumer.receive(1000);
		if(message == null)
		{
			System.out.println("message is null");
			return null;
		}
		int length = message.getContent().length;
		byte[] bytes = new byte[length];
		message.readBytes(bytes);
		
		consumer.close();
		
		return bytes;
	}
}
